package com.danbay.framework.kafka;

import com.danbay.framework.exception.BusinessException;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author SJG
 *         2018/1/13.
 */
public class KafkaContext {

    private static final ThreadLocal<String> localToken = new ThreadLocal<>();
    private static final ThreadLocal<String> localTopic = new ThreadLocal<>();
    private static final ConcurrentMap<String, String> businessTopicMap = new ConcurrentHashMap<>();

    /**
     * 存放当前线程kafka消息中的token
     *
     * @param token token
     */
    public static void putToken(String token) {
        localToken.set(token);
    }

    /**
     * 取出存放在当前线程中kafka消息中的token
     *
     * @return token
     */
    public static String popToken() {
        String token = localToken.get();
        localToken.remove();
        return token;
    }

    /**
     * 收到消息时设置保存当前的主题名
     *
     * @param topic 主题名
     */
    public static void putTopic(String topic) {
        localTopic.set(topic);
    }

    /**
     * 取出存放在当前线程中kafka的topic
     *
     * @return topic
     */
    public static String popTopic() {
        String topic = localTopic.get();
        localTopic.remove();
        return topic;
    }

    /**
     * 获取放在当前线程中kafka的topic
     *
     * @return topic
     */
    public static String getTopic() {
        return localTopic.get();
    }


    /**
     * 根据业务key与topic进行映射
     *
     * @param key 业务key
     */
    public static void mapTopic(String key) {
        businessTopicMap.put(key, getTopic());
    }

    /**
     * 根据业务key与topic进行映射
     *
     * @param key   业务key
     * @param topic 主题
     */
    public static void mapTopic(String key, String topic) {
        businessTopicMap.put(key, topic);
    }

    /**
     * 使用业务key获取topic实例
     * key为调用mapTopic时添加的key
     *
     * @param kafkaClient kafkaClient
     * @param key         业务key
     * @return AccessKafkaTopic
     */
    public static AccessKafkaTopic requestTopic(KafkaClient kafkaClient, String key) throws BusinessException {
        if (!businessTopicMap.containsKey(key)) {
            throw new BusinessException(key + "未找到对应的topic");
        }
        String topic = businessTopicMap.get(key);
        return kafkaClient.requestTopic(topic);
    }
}
